package com.lzy.canal.component.consumer.mysql;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.lzy.canal.component.consumer.MysqlConsumer;
import com.lzy.canal.component.result.mysql.DefaultConsumerDmlRestData;
import com.lzy.canal.component.row.CanalRow;
import com.lzy.canal.util.CommonUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * 处理正常的
 *
 */
@Slf4j
public class DefaultDmlConsumer implements MysqlConsumer<DefaultConsumerDmlRestData> {

    @Override
    public List<DefaultConsumerDmlRestData> consumer(CanalRow row) {
        CanalEntry.RowChange rowChage = getRowChage(row);
        return rowChage.getRowDatasList().stream()
                .map(o -> defaultDmlConsumerRestData(o, rowChage.getEventType()))
                .collect(Collectors.toList());
    }

    /**
     * mysql 一行消息数据转换成es数据
     *
     * @param rowData   记录行
     * @param eventType 操作类型
     * @return
     */
    protected DefaultConsumerDmlRestData defaultDmlConsumerRestData(CanalEntry.RowData rowData, CanalEntry.EventType eventType) {
        List<CanalEntry.Column> columns = getColumn(rowData, eventType);
        Map<String, Object> map = new HashMap<>(columns.size());
        for (CanalEntry.Column column : columns) {
            if (column.getIsNull()) {
                map.put(column.getName(), null);
                continue;
            }
            map.put(column.getName(), CommonUtil.convertValueType(column));
        }
        return new DefaultConsumerDmlRestData(map, eventType);
    }


}
